package com.dja.nzw.mq.kafka.simple;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * 本地测试会出现无法连接到kafka服务的异常，重试3次后异常退出
 * 解决1：
 *  metadata.broker.list 这个参数设置为kafka broker所在的主机名，并且需要在hosts做映射
 *
 * 解决2：
 *  修改kafka启动的配置参数advertised.host.name ，将其设置为kafka broker所在的的主机ip，这样producer中metadata.broker.list的配置可以设置为IP的形式
 *
 *  究其原因，producer和consumer是通过主机名（advertised.host.name）来连接broker的，而如果这个值没有设置，则会默认使用kafka broker当前所在的主机名（即java.net.InetAddress.getCanonicalHostName()）
 * Created by shuyun on 2017/7/12.
 */
public class SimpleProducer {

    public static void main(String[] args) {
        String topic = "dja-topic";
        Producer<String, String> producer = null;

        try {
            Properties props = new Properties();
            props.put("metadata.broker.list", "cs03:9092");
            props.put("serializer.class", "kafka.serializer.StringEncoder");
//            props.put("partitioner.class", "com.dja.nzw.mq.kafka.simple.SimplePartitioner");
            props.put("request.required.acks", "1");

            ProducerConfig config = new ProducerConfig(props);
            producer = new Producer<String, String>(config);

            long events = 10l;
            Random rnd = new Random();
            for (long nEvents = 0; nEvents < events; nEvents++) {
                long runtime = new Date().getTime();
                String ip = "192.168.2." + rnd.nextInt(255);
                String msg = runtime + ",www.example.com," + ip;
                KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, ip, msg);
                producer.send(data);

                System.out.println("消息发送成功：key=" + ip + "; message=" + msg);
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            System.exit(-1);
        } finally {
            if(null != producer)
                producer.close();
        }
    }
}
